package org.dromara.system.config.core;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.base.Throwables;
import lombok.extern.slf4j.Slf4j;
import org.dromara.system.config.redis.RedisDelayMQMessage;
import org.dromara.system.config.redis.RedisMQListenerTarget;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author lilu
 */
@Slf4j
@Component
public class RedisDelayMQRegister implements ApplicationRunner, ApplicationContextAware, DisposableBean {

    private static final String THREAD_PREFIX = "redis-delay-mq-thread-";

    private final Set<String> registerQueueName = new HashSet<>();

    private final List<Worker> registerQueueThreads = new ArrayList<>();

    private final RedisTemplate redisTemplate;

    private ApplicationContext applicationContext;

    public RedisDelayMQRegister(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    @Override
    public void destroy() throws Exception {
        if (CollUtil.isNotEmpty(registerQueueThreads)) {
            registerQueueThreads.forEach(worker -> worker.exit());
        }
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        initializer();
        for (String queueName : registerQueueName) {
            Worker worker = new Worker(queueName);
            worker.setName(THREAD_PREFIX + queueName);
            registerQueueThreads.add(worker);
            worker.start();
            log.info("start delay message queue listener [{}]", queueName);
        }
    }

    private void initializer() {
        List<RedisMQListenerTarget> redisMQListenerTargets = RedisDelayMQListenerScanner.getRedisListenerTargets();
        if (CollUtil.isNotEmpty(redisMQListenerTargets)) {
            for (RedisMQListenerTarget redisMQListenerTarget : redisMQListenerTargets) {
                registerQueueName.add(redisMQListenerTarget.getQueueName());
            }
        }
    }

    private class Worker extends Thread {

        private boolean exit = false;

        private final String queueName;

        private List<RedisMQListenerTarget> consumerTargets = new ArrayList<>();

        private Worker(String queueName) {
            this.queueName = queueName;
            initConsumerTargets();
        }

        private void initConsumerTargets() {
            if (consumerTargets.size() == 0) {
                List<RedisMQListenerTarget> redisMQListenerTargets = RedisDelayMQListenerScanner.getRedisListenerTargets();
                consumerTargets.addAll(redisMQListenerTargets.stream().filter(f -> f.match(queueName)).collect(Collectors.toList()));
            }
        }

        @Override
        public void run() {
            if (StrUtil.isBlank(queueName)) {
                return;
            }
            while (!exit) {
                Set<Object> set = redisTemplate.opsForZSet().rangeByScore(queueName, 0, System.currentTimeMillis(), 0, 1);
                if (CollUtil.isEmpty(set)) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(100);
                    } catch (InterruptedException e) {
                        log.error("sleep exception, error=>{}", Throwables.getStackTraceAsString(e));
                    }
                } else {
                    Object message = null;
                    try {
                        message = set.iterator().next();
                        if (clearMessage(message)) {
                            handleMessage((RedisDelayMQMessage) message);
                        }
                    } catch (IllegalAccessException | InvocationTargetException e) {
                        recoverMessage((RedisDelayMQMessage) message);
                        log.error("delay message queue [{}] exception to handle message, error=>{}", queueName, Throwables.getStackTraceAsString(e));
                    }
                }
            }
        }

        private void handleMessage(RedisDelayMQMessage message) throws IllegalAccessException, InvocationTargetException {
            if (consumerTargets.size() > 0) {
                for (RedisMQListenerTarget target : consumerTargets) {
                    Method targetMethod = target.getMethod();
                    if (target.getMethodParameterClassName().equals(RedisDelayMQMessage.class.getName())) {
                        targetMethod.invoke(target.getBean(applicationContext), message);
                    } else if (target.getMethodParameterClassName().equalsIgnoreCase(message.getPayload().getClass().getName())) {
                        targetMethod.invoke(target.getBean(applicationContext), message.getPayload());
                    } else {
                        throw new RedisMQException(StrUtil.format("the message type in the delay message queue [{}] is inconsistent with the message type defined in the [{}] method", queueName, targetMethod.getName()));
                    }
                }
            } else {
                throw new RedisMQException(StrUtil.format("the message type in the delay message queue [{}], there is no corresponding target consumer", queueName));
            }
        }

        private boolean clearMessage(Object message) {
            return redisTemplate.opsForZSet().remove(queueName, message) > 0;
        }

        private void recoverMessage(RedisDelayMQMessage message) {
            if (Objects.isNull(message)) {
                return;
            }
            // 判断是否恢复
            boolean recover = false;
            Integer failRepeatNumber = message.getFailRepeatNumber();
            if (failRepeatNumber > 0) {
                if (message.getFailNumber() < failRepeatNumber) {
                    recover = true;
                }
            } else if (failRepeatNumber < 0) {
                recover = true;
            }
            if (recover) {
                message.setFailNumber(message.getFailNumber() + 1);
                redisTemplate.opsForZSet().add(queueName, message, System.currentTimeMillis() + message.getFailRepeatInterval());
            }
        }

        public void exit() {
            this.exit = true;
            log.error("delay message queue [{}] stops listening", queueName);
        }

    }

}
